package com.xwintop.xTransfer.receiver.service.impl;

import com.google.common.base.Strings;
import com.xwintop.xTransfer.common.MsgLogger;
import com.xwintop.xTransfer.common.model.LOGKEYS;
import com.xwintop.xTransfer.common.model.LOGVALUES;
import com.xwintop.xTransfer.common.model.Msg;
import com.xwintop.xTransfer.messaging.*;
import com.xwintop.xTransfer.receiver.bean.ReceiverConfig;
import com.xwintop.xTransfer.receiver.bean.ReceiverConfigPulsar;
import com.xwintop.xTransfer.receiver.service.Receiver;
import com.xwintop.xTransfer.task.quartz.TaskQuartzJob;
import com.xwintop.xcore.util.UuidUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.*;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;

import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: ReceiverPulsarImpl
 * @Description: Pulsar接收器实现类
 * @author: xufeng
 * @date: 2021/12/12 17:53
 */

@Service("receiverPulsar")
@Scope("prototype")
@Slf4j
public class ReceiverPulsarImpl implements Receiver {
    private ReceiverConfigPulsar receiverConfigPulsar;
    private MessageHandler messageHandler;

    private Consumer consumer = null;
    private PulsarClient pulsarClient = null;

    @Override
    public void receive(Map params) throws Exception {
        this.initPulsarClient();
        synchronized (this) {
            if (consumer == null) {
                ConsumerBuilder<byte[]> consumerBuilder = this.pulsarClient.newConsumer(Schema.BYTES);
                consumerBuilder.consumerName(receiverConfigPulsar.getConsumerName());
                consumerBuilder.subscriptionName(receiverConfigPulsar.getSubscriptionName());
                consumerBuilder.topic("persistent://" + receiverConfigPulsar.getTenant() + "/" + receiverConfigPulsar.getNamespace() + "/" + receiverConfigPulsar.getTopic());
                consumerBuilder.subscriptionType(SubscriptionType.valueOf(receiverConfigPulsar.getSubscriptionType()));
                consumerBuilder.messageListener((consumer, message) -> {
                    try {
                        log.debug("接收到消息：" + message.getMessageId());
                        IMessage msg = new DefaultMessage();
                        msg.setRawData(message.getValue());
                        msg.setProperty(LOGKEYS.CHANNEL_IN_TYPE, LOGVALUES.CHANNEL_TYPE_PULSAR);
                        msg.setProperty(LOGKEYS.CHANNEL_IN, receiverConfigPulsar.getServiceUrl() + ":" + receiverConfigPulsar.getTopic());
                        msg.setProperty(LOGKEYS.RECEIVER_TYPE, LOGVALUES.CHANNEL_TYPE_PULSAR);
                        msg.setProperty(LOGKEYS.RECEIVER_ID, receiverConfigPulsar.getId());
                        msg.setFileName(StringUtils.defaultIfEmpty(message.getProperty(receiverConfigPulsar.getFileNameField()), UuidUtil.get32UUID()));
                        msg.getProperties().putAll(message.getProperties());
                        IContext ctx = new DefaultContext();
                        ctx.setMessage(msg);

                        Msg msgLogInfo = new Msg(LOGVALUES.EVENT_MSG_RECEIVED, msg.getId(), null);
                        msgLogInfo.put(LOGKEYS.CHANNEL_IN_TYPE, LOGVALUES.CHANNEL_TYPE_PULSAR);
                        msgLogInfo.put(LOGKEYS.CHANNEL_IN, receiverConfigPulsar.getServiceUrl() + ":" + receiverConfigPulsar.getTopic());
                        msgLogInfo.put(LOGKEYS.MSG_TAG, msg.getFileName());
                        msgLogInfo.put(LOGKEYS.MSG_LENGTH, ArrayUtils.getLength(msg.getMessage()));
                        msgLogInfo.put(LOGKEYS.JOB_ID, params.get(TaskQuartzJob.JOBID));
                        msgLogInfo.put(LOGKEYS.JOB_SEQ, params.get(TaskQuartzJob.JOBSEQ));
                        msgLogInfo.put(LOGKEYS.RECEIVER_TYPE, LOGVALUES.RCV_TYPE_MQ);
                        msgLogInfo.put(LOGKEYS.RECEIVER_ID, receiverConfigPulsar.getId());
                        MsgLogger.info(msgLogInfo.toMap());
                        messageHandler.handle(ctx);

                        consumer.acknowledge(message);
                    } catch (Exception e) {
                        log.error("{} 处理Pulsar接收消息失败：", receiverConfigPulsar.getId(), e);
                        consumer.negativeAcknowledge(message);
                    }
                });
                if (receiverConfigPulsar.getAckTimeoutMs() > 0) {
                    consumerBuilder.ackTimeout(receiverConfigPulsar.getAckTimeoutMs(), TimeUnit.MILLISECONDS);
                }
                DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterBuilder = null;
                if (receiverConfigPulsar.getDeadLetterPolicyMaxRedeliverCount() >= 0) {
                    deadLetterBuilder = DeadLetterPolicy.builder().maxRedeliverCount(receiverConfigPulsar.getDeadLetterPolicyMaxRedeliverCount());
                }
                if (receiverConfigPulsar.getMaxRedeliverCount() >= 0) {
                    deadLetterBuilder = DeadLetterPolicy.builder().maxRedeliverCount(receiverConfigPulsar.getMaxRedeliverCount());
                }
                if (deadLetterBuilder != null && StringUtils.isNotBlank(receiverConfigPulsar.getDeadLetterTopic())) {
                    deadLetterBuilder.deadLetterTopic("persistent://" + receiverConfigPulsar.getTenant() + "/" + receiverConfigPulsar.getNamespace() + "/" + receiverConfigPulsar.getDeadLetterTopic());
                }
                if (deadLetterBuilder != null) {
                    consumerBuilder.deadLetterPolicy(deadLetterBuilder.build());
                }
                consumer = consumerBuilder.subscribe();
                log.info("{} ReceiverPulsarImpl poll start...", receiverConfigPulsar.getId());
            }
        }
    }

    @Override
    public void setReceiverConfig(ReceiverConfig receiverConfig) {
        this.receiverConfigPulsar = (ReceiverConfigPulsar) receiverConfig;
    }

    @Override
    public void setMessageHandler(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }

    @Override
    public void destroy() {
        if (consumer != null) {
            try {
                consumer.close();
            } catch (Exception e) {
                log.error("{} destroy异常：", receiverConfigPulsar.getId(), e);
            }
            consumer = null;
        }
        if (pulsarClient != null) {
            try {
                pulsarClient.close();
            } catch (Exception e) {
                log.error("{} destroy异常：", receiverConfigPulsar.getId(), e);
            }
            pulsarClient = null;
        }
    }

    public synchronized void initPulsarClient() throws PulsarClientException {
        if (pulsarClient == null) {
            ClientBuilder pulsarClientBuilder = PulsarClient.builder();
            pulsarClientBuilder.serviceUrl(receiverConfigPulsar.getServiceUrl());
            pulsarClientBuilder.ioThreads(receiverConfigPulsar.getIoThreads());
            pulsarClientBuilder.listenerThreads(receiverConfigPulsar.getListenerThreads());
            pulsarClientBuilder.enableTcpNoDelay(receiverConfigPulsar.isEnableTcpNoDelay());
            pulsarClientBuilder.keepAliveInterval(receiverConfigPulsar.getKeepAliveIntervalSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.connectionTimeout(receiverConfigPulsar.getConnectionTimeoutSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.operationTimeout(receiverConfigPulsar.getOperationTimeoutSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.startingBackoffInterval(receiverConfigPulsar.getStartingBackoffIntervalMs(), TimeUnit.MILLISECONDS);
            pulsarClientBuilder.maxBackoffInterval(receiverConfigPulsar.getMaxBackoffIntervalSec(), TimeUnit.SECONDS);
            pulsarClientBuilder.useKeyStoreTls(receiverConfigPulsar.isUseKeyStoreTls());
            pulsarClientBuilder.tlsTrustCertsFilePath(receiverConfigPulsar.getTlsTrustCertsFilePath());
            pulsarClientBuilder.tlsCiphers(new HashSet<>(receiverConfigPulsar.getTlsCiphers()));
            pulsarClientBuilder.tlsProtocols(new HashSet<>(receiverConfigPulsar.getTlsProtocols()));
            pulsarClientBuilder.tlsTrustStorePassword(receiverConfigPulsar.getTlsTrustStorePassword());
            pulsarClientBuilder.tlsTrustStorePath(receiverConfigPulsar.getTlsTrustStorePath());
            pulsarClientBuilder.tlsTrustStoreType(receiverConfigPulsar.getTlsTrustStoreType());
            pulsarClientBuilder.allowTlsInsecureConnection(receiverConfigPulsar.isAllowTlsInsecureConnection());
            pulsarClientBuilder.enableTlsHostnameVerification(receiverConfigPulsar.isEnableTlsHostnameVerification());
            if (!Strings.isNullOrEmpty(receiverConfigPulsar.getTlsAuthCertFilePath()) && !Strings.isNullOrEmpty(receiverConfigPulsar.getTlsAuthKeyFilePath())) {
                pulsarClientBuilder.authentication(AuthenticationFactory.TLS(receiverConfigPulsar.getTlsAuthCertFilePath(), receiverConfigPulsar.getTlsAuthKeyFilePath()));
            }

            if (!Strings.isNullOrEmpty(receiverConfigPulsar.getTokenAuthValue())) {
                pulsarClientBuilder.authentication(AuthenticationFactory.token(receiverConfigPulsar.getTokenAuthValue()));
            }
            pulsarClient = pulsarClientBuilder.build();
        }
    }
}
